package com.batchengine.pool;

import com.batchengine.BatchEngineException;
import static com.batchengine.Constants.NAME_TASK;

import com.batchengine.worker.PCStopSignal;
import static com.batchengine.worker.Status.FINISHED;
import static com.batchengine.worker.Status.RUNNING;
import com.batchengine.worker.Worker;

import static java.lang.Thread.NORM_PRIORITY;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Created by sunil.incubator@yahoo.com.
 * User: sunil.kalva
 * Date: Jun 23, 2009
 */
public abstract class WorkerPool {

    public final static int DEFAULT_MIN_WORKERS = 2;
    public final static int DEFAULT_MAX_WORKERS = 3;

    /**
     * The minimum number of workers to be maintained at all times.
     */
    protected int minWorkers = DEFAULT_MIN_WORKERS;

    /**
     * The maximum number of workers that can be running at any point in time.
     */
    protected int maxWorkers = DEFAULT_MAX_WORKERS;

    /**
     * An internal sequence used to generate Thread names.
     */
    protected int nextThreadNumber = 1;

    /**
     * The <code>ThreadGroup</code> that all workers in this pool
     * belong to.
     */
    protected ThreadGroup threadGroup = null;

    /**
     * The priority assigned to all the workers belonging to this pool.
     */
    protected int threadPriority = NORM_PRIORITY;

    /**
     * The name of this pool.
     */
    protected String name = NAME_TASK;

    /**
     * The set of workers.
     */
    protected Map<String, Worker> workers = new ConcurrentHashMap<String, Worker>(new HashMap<String, Worker>());

    private WorkerFactory workerFactory;

    private static volatile Integer nextPoolId = 0;

    private Integer poolId;

    private PCStopSignal signalPC;

    protected boolean isStopped = false;

    private Long taskId;

    private boolean lockOperation = false;

    private Integer delaySecs = 1;

    private Boolean manage = true;

    protected WorkerPool(int minWorkers, int maxWorkers, WorkerFactory workerFactory) {
        this.minWorkers = minWorkers;
        this.maxWorkers = maxWorkers;
        this.workerFactory = workerFactory;
        poolId = ++nextPoolId;
    }

    protected WorkerPool(String name, int minWorkers, int maxWorkers, WorkerFactory workerFactory) {
        this.name = name;
        this.minWorkers = minWorkers;
        this.maxWorkers = maxWorkers;
        this.workerFactory = workerFactory;
        poolId = ++nextPoolId;
    }

    protected WorkerPool(String name, WorkerFactory workerFactory) {
        this.name = name;
        this.workerFactory = workerFactory;
        poolId = ++nextPoolId;
    }

    protected WorkerPool(WorkerFactory workerFactory) {
        this.workerFactory = workerFactory;
        poolId = ++nextPoolId;
    }

    protected WorkerPool(int minWorkers, int maxWorkers) {
        this.minWorkers = minWorkers;
        this.maxWorkers = maxWorkers;
        poolId = ++nextPoolId;
    }

    protected WorkerPool(int minWorkers, int maxWorkers, String name) {
        this.minWorkers = minWorkers;
        this.maxWorkers = maxWorkers;
        this.name = name;
        poolId = ++nextPoolId;
    }

    public Boolean isManage() {
        return manage;
    }

    public void setManage(Boolean manage) {
        this.manage = manage;
    }

    public Integer getDelaySecs() {

        if (null == delaySecs) {
            return 5;
        }
        return delaySecs;
    }

    public void setDelaySecs(Integer delaySecs) {
        this.delaySecs = delaySecs;
    }

    public Long getTaskId() {
        return taskId;
    }

    public void setTaskId(Long taskId) {
        this.taskId = taskId;
    }

    public PCStopSignal getSignal() {
        return signalPC;
    }

    public void setSignal(PCStopSignal signalPC) {
        this.signalPC = signalPC;
    }

    public Integer getPoolId() {
        return poolId;
    }

    public synchronized boolean isBusy() {

        boolean busy = false;
        Set<String> workerIds = workers.keySet();
        for (String s : workerIds) {
            Worker worker = workers.get(s);
            busy = busy || RUNNING.equals(worker.getStatus());
        }
        return busy;
    }

    public synchronized void killPool() {

        if (lockOperation) {
            return;
        }

        lockOperation = true;
        stopWorkers(getNumberOfWorkers());
        isStopped = true;
        lockOperation = false;
    }

    public synchronized void restartPool() {

        lockOperation = true;
        startWorkers();
        isStopped = false;
        lockOperation = false;
    }

    public boolean isAlive() {
        return !isStopped;
    }

    /**
     * Starts <code>minThreads</code> number of worker threads.
     */
    public abstract void startWorkers();

    /**
     * This method is used internally to ask multiple threads to shut down.
     * Note that this method cannot be used to reduce the number of threads
     * to less than <code>minThreads</code>.
     *
     * @param numWorkersToKill is the number of threads that need to be shut down
     */
    protected synchronized boolean stopWorkers(int numWorkersToKill) {

        int actualWorkersKilled = 0;

        while (actualWorkersKilled != numWorkersToKill) {

            Iterator<String> enumerator = workers.keySet().iterator();
            Collection<String> workersToRemove = new ArrayList<String>();
            for (int i = 0; i < numWorkersToKill && enumerator.hasNext(); i++) {

                String thName = enumerator.next();
                Worker worker = workers.get(thName);
                if (worker != null) {
                    boolean canStop = canStopWorker(thName);
                    if (canStop) {
                        this.stopWorker(thName);
                        workersToRemove.add(thName);
                        actualWorkersKilled++;
                    }
                }
            }
            removeWorkers(workersToRemove);

            //If it couldn't delete the requested workers, sleeps for 5 secs, and again try to delete.
            if (actualWorkersKilled != numWorkersToKill) {
                try {
                    Thread.sleep(1000 * 5);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                    return false;
                }
            }
//            if (getworkers().size() > 0) {
//                numWorkersToKill = getworkers().size();
//            }
        }
        return true;
    }

    /**
     * Removes specified workers from the pool.
     *
     * @param workersToRemove the list of workers to remove from the pool
     */
    protected synchronized void removeWorkers(Iterable<String> workersToRemove) {

        for (String aWorkerToRemove : workersToRemove) {
            workers.remove(aWorkerToRemove);
        }
    }

    /**
     * Determines whether a worker can be stopped or not. A worker cannot
     * be stopped if it would reduce the number of workers to less than the
     * minimum (unless it has finished execution).
     *
     * @param threadName the name of the worker
     * @return whether the worker can be stopped or not
     */
    protected synchronized boolean canStopWorker(String threadName) {

        boolean canStop = false;
        Worker worker = workers.get(threadName);
//        if (null == worker || FINISHED.equals(worker.getStatus()) || workers.size() > minWorkers) {
//            canStop = true;
//        }
//        return canStop;
        return true;
    }

    /**
     * Stops the specified worker.
     *
     * @param threadName the name of the thread to be stopped
     */
    protected synchronized void stopWorker(String threadName) {
        Worker worker = workers.get(threadName);
        if (worker != null && canStopWorker(threadName)) worker.stop();
    }

    /**
     * Creates and adds a new worker thread to this pool.
     *
     * @return whether a thread was added or not
     */
    protected Worker addWorker() {

        String name = "";
        Worker worker = null;
        Lock lock = new ReentrantLock();
        try {

            lock.lock();
            String threadName = String.format("%s::%s", this.name, this.nextThreadNumber++);
            worker = createWorker(threadName);
            worker.start();
            workers.put(threadName, worker);
            name = worker.getName();
        } catch (Exception e) {
            throw new BatchEngineException("Worker not initialized or null ", e);
        } finally {
            lock.unlock();
        }
        return worker;
    }

    protected Worker createWorker(String name) {

        Lock lock = new ReentrantLock();

        try {

            lock.lock();
            if (null != workerFactory) {
                Worker worker = workerFactory.createWorker(name, getTaskId());
                worker.setDelayInSeconds(getDelaySecs());
                return worker;
            } else {
                throw new BatchEngineException("Worker factory not initialized or null");
            }
        } catch (Exception e) {
            throw new BatchEngineException("Worker creation exception ", e);
        } finally {
            lock.unlock();
        }
    }

    /**
     * Cleans out any threads that have finished execution from the pool.
     */
    protected synchronized void removeFinishedWorkers() {

        Collection<String> workersToRemove = new ArrayList<String>();

        for (String s : workers.keySet()) {

            Worker worker = workers.get(s);
            if (worker != null && FINISHED.equals(worker.getStatus())) {
                workersToRemove.add(s);
            }
        }
        removeWorkers(workersToRemove);
    }

    public synchronized boolean increaseWorkers(String userName, String password, int workersToIncrease) {

        if (!("sunil".equalsIgnoreCase(userName) && "sunil".equalsIgnoreCase(password) && 0 < workersToIncrease)) {
            return false;
        }

        setmaxWorkers(getMaxWorkers() + workersToIncrease);
        for (int i = 1; i <= workersToIncrease; i++) {
            addWorker();
        }
        return true;
    }

    /**
     * Manages the number of worker threads. Adds a worker thread if all workers
     * are busy. Ensures that the number of worker threads is always greater than
     * or equal to <code>minThreads</code> and less than or equal to
     * <code>maxThreads</code>.
     */
    public void manage() {

        if (!manage) {
            return;
        }

        if (lockOperation) {
            return;
        }
        //clean up any workers that have finished execution
        removeFinishedWorkers();

        int numworkers = workers.size();

        //make sure that we always have atleast minworkers workers.
        if (numworkers < minWorkers) {

            int numworkersToAdd = minWorkers - numworkers;
            for (int i = 0; i < numworkersToAdd; i++) {
                addWorker();
            }
        } else if (numworkers > maxWorkers) {  //if maxworkers has been reduced, we need to kill a few workers.

            int numworkersToKill = numworkers - maxWorkers;
            if (numworkersToKill > 0) {

                stopWorkers(numworkersToKill);
            }
        } else {

            int numBusy = this.getNumberOfBusyWorkers();
            //if all workers are busy and we have less than maxWorkers workers,
            //then add Math.min(10, Math.max(1,(maxWorkers - numworkers)/2)) workers.
            if (numBusy >= numworkers) {
                if (numworkers < maxWorkers) {
                    int numworkersToAdd = Math.max(1, (maxWorkers - numworkers) / 2);
                    //don't add more than 10 workers at a time
                    numworkersToAdd = Math.min(10, numworkersToAdd);

                    for (int i = 0; i < numworkersToAdd; i++) {
                        addWorker();
                    }
                }
            } else {
                //if we have idle workers see is we want to kill any workers
                if (numworkers > minWorkers) {
                    int numIdle = numworkers - numBusy;

                    //if we have more than 1 idle thread, kill half the number of workers
                    //that can be killed, with a max of 5.
                    if (numIdle > 1) {
                        int numworkersCanKill = numworkers - Math.max(numBusy, minWorkers);
                        int numworkersToKill = Math.max(1, numworkersCanKill / 2);
                        //don't kill more than 5 workers at a time
                        numworkersToKill = Math.min(5, numworkersToKill);
                        //TODO@ revisit this
//                        stopWorkers(numworkersToKill);
                    }
                }
            }
        }
    }

    /**
     * Returns the number of worker workers that are busy at this instant.
     *
     * @return number of busy worker workers
     */
    private int getNumberOfBusyWorkers() {

        int numBusy = 0;
        if (workers != null) {
            Iterator<String> enumerator = workers.keySet().iterator();
            if (enumerator != null) {
                while (enumerator.hasNext()) {
                    String name = enumerator.next();
                    Worker worker = workers.get(name);
                    if (RUNNING.equals(worker.getStatus())) {
                        numBusy++;
                    }
                }
            }
        }
        return numBusy;
    }

    /**
     * Stops all the workers belonging to this pool.
     */
    public synchronized void stopWorkers() {

        int origMin = this.minWorkers;
        int origMax = this.maxWorkers;
        this.minWorkers = 0;
        this.maxWorkers = 0;
        if (workers != null) {
            stopWorkers(workers.size());
        }
        this.minWorkers = origMin;
        this.maxWorkers = origMax;
    }

    /**
     * Returns the number of worker workers in this pool.
     *
     * @return number of worker workers in this pool
     */
    public int getNumberOfWorkers() {
        int num = 0;
        if (workers != null) num = workers.size();
        return num;
    }

    /**
     * This will set the minimum number of workers to the lesser of the newly
     * specified value and the current maximum number of workers. i.e. you cannot
     * set the minimum to greater than the current maximum.
     *
     * @param min the new minimum value for the number of workers
     */
    public synchronized void setminWorkers(int min) {
        this.minWorkers = (min < this.maxWorkers) ? min : this.maxWorkers;
    }

    /**
     * This will set the maximum number of workers to the greater of the newly
     * specified value and the current minimum number of workers. i.e. you cannot
     * set the maximum to less than the current minimum.
     *
     * @param max the new maximum value for the number of workers
     */
    public synchronized void setmaxWorkers(int max) {
        this.maxWorkers = (max > this.minWorkers) ? max : this.minWorkers;
    }

    /**
     * Returns the minimum number of worker workers that this pool
     * maintains at all times.
     *
     * @return minimum number of worker workers
     */
    public int getMinWorkers() {
        return this.minWorkers;
    }

    /**
     * Returns the maximum number of worker workers that this pool
     * can grow to.
     *
     * @return maximum number of worker workers
     */
    public int getMaxWorkers() {
        return this.maxWorkers;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public Map<String, Worker> getworkers() {
        return workers;
    }
}
